import os
import asyncio
import requests
import soundfile as sf
import numpy as np
import time
from utils.logger import setup_logger
from typing import Dict

logger = setup_logger("ai.tts")
import os
import asyncio
import requests
import soundfile as sf
import numpy as np
import time
from utils.logger import setup_logger
from typing import Dict

logger = setup_logger("ai.tts")


class TTSEngine:
    async def generate_audio_and_play(self, text: str):
        # 全局TTS开关判断
        if hasattr(self, 'config') and not self.config['tts'].get('enable', True):
            logger.info("TTS全局开关关闭，跳过语音合成和播放")
            return
        # ...原有逻辑...

    def _cleanup_cache(self):
        """清理缓存目录，超出最大数量时删除最久未用的音频文件"""
        max_files = 30
        try:
            if hasattr(self, 'audio_cache_max_files'):
                max_files = int(self.audio_cache_max_files)
            elif hasattr(self, 'config') and 'audio_cache_max_files' in self.config.get('tts', {}):
                max_files = int(self.config['tts']['audio_cache_max_files'])
        except Exception:
            pass
        files = [os.path.join(self.cache_dir, f) for f in os.listdir(self.cache_dir) if f.endswith('.wav')]
        if len(files) > max_files:
            files.sort(key=lambda x: os.path.getmtime(x))
            to_delete = files[:len(files)-max_files]
            for f in to_delete:
                try:
                    os.remove(f)
                    logger.info(f"自动清理缓存音频: {f}")
                except Exception as e:
                    logger.warning(f"删除缓存文件失败: {f}, {e}")

    def get_engine(self):
        # 支持简洁写法
        val = getattr(self, 'engine', 'gpt-sovits').lower()
        if val in ('gpt-sovits', 'sovits'):
            return 'gpt-sovits'
        if val in ('vits', 'vits-simple-api'):
            return 'vits-simple-api'
        if val in ('huahua', 'hua-hua-api'):
            return 'hua-hua-api'
        if val == 'custom':
            return 'custom'
        return 'gpt-sovits'

    async def hua_hua_api_generate_audio(self, text: str) -> tuple[np.ndarray, str]:
        if hasattr(self, 'config') and not self.config['tts'].get('enable', True):
            logger.info("TTS全局开关关闭，跳过huahua接口")
            empty_audio = np.zeros(22050, dtype=np.float32)
            return empty_audio, "empty.wav"
        """使用huahua接口生成语音音频"""
        # logger.debug(f"[TTS调用] hua_hua_api_generate_audio 参数 text: {repr(text)}")
        if not text:
            logger.warning("空文本，生成静音音频")
            empty_audio = np.zeros(22050, dtype=np.float32)
            return empty_audio, "empty.wav"

        cache_file = os.path.join(self.cache_dir, f"huahua_{hash(text)}.wav")
        if os.path.exists(cache_file):
            logger.info(f"[huahua_api] 使用缓存音频: {cache_file}")
            try:
                data, samplerate = sf.read(cache_file)
                return data, cache_file
            except Exception as e:
                logger.error(f"[huahua_api] 读取缓存音频失败: {e}")

        try:
            url = getattr(self, 'huahua_url', None) or self.config['tts'].get('huahua_url', '')
            if not url or not url.startswith('http'):
                logger.error("[huahua_api] TTS接口URL未配置或非法，请在配置中填写正确的huahua_url！")
                empty_audio = np.zeros(22050, dtype=np.float32)
                return empty_audio, "error.wav"
            params = {
                "text": text,
                "language": "zh",
                "top_k": 15,
                "top_p": 1.0,
                "temperature": 1.0
            }
            start_time = time.time()
            response = await asyncio.to_thread(
                requests.get,
                url,
                params=params,
                timeout=30
            )
            if response.status_code == 200:
                with open(cache_file, "wb") as f:
                    f.write(response.content)
                data, samplerate = sf.read(cache_file)
                logger.info(f"[huahua_api] 音频生成成功，耗时: {time.time() - start_time:.2f}s")
                return data, cache_file
            else:
                logger.error(f"[huahua_api] TTS服务错误: {response.status_code} - {response.text}")
                empty_audio = np.zeros(22050, dtype=np.float32)
                return empty_audio, "error.wav"
        except Exception as e:
            logger.error(f"[huahua_api] TTS调用异常: {e}", exc_info=True)
            empty_audio = np.zeros(22050, dtype=np.float32)
            return empty_audio, "error.wav"

    async def custom_api_generate_audio(self, text: str) -> tuple[np.ndarray, str]:
        if hasattr(self, 'config') and not self.config['tts'].get('enable', True):
            logger.info("TTS全局开关关闭，跳过custom接口")
            empty_audio = np.zeros(22050, dtype=np.float32)
            return empty_audio, "empty.wav"
        """使用自定义接口生成语音音频，支持url和params模板"""
        # logger.debug(f"[TTS调用] custom_api_generate_audio 参数 text: {repr(text)}")
        if not text or not hasattr(self, 'custom_url'):
            logger.warning("空文本或未配置自定义接口，生成静音音频")
            empty_audio = np.zeros(22050, dtype=np.float32)
            return empty_audio, "empty.wav"

        cache_file = os.path.join(self.cache_dir, f"custom_{hash(text)}.wav")
        if os.path.exists(cache_file):
            logger.info(f"[custom_api] 使用缓存音频: {cache_file}")
            try:
                data, samplerate = sf.read(cache_file)
                return data, cache_file
            except Exception as e:
                logger.error(f"[custom_api] 读取缓存音频失败: {e}")

        try:
            url = self.custom_url.replace("{text}", text)
            if not url or not url.startswith('http'):
                logger.error("[custom_api] TTS接口URL未配置或非法，请在配置中填写正确的custom_url！")
                empty_audio = np.zeros(22050, dtype=np.float32)
                return empty_audio, "error.wav"
            start_time = time.time()
            response = await asyncio.to_thread(
                requests.get,
                url,
                timeout=30
            )
            if response.status_code == 200:
                with open(cache_file, "wb") as f:
                    f.write(response.content)
                data, samplerate = sf.read(cache_file)
                logger.info(f"[custom_api] 音频生成成功，耗时: {time.time() - start_time:.2f}s")
                return data, cache_file
            else:
                logger.error(f"[custom_api] TTS服务错误: {response.status_code} - {response.text}")
                empty_audio = np.zeros(22050, dtype=np.float32)
                return empty_audio, "error.wav"
        except Exception as e:
            logger.error(f"[custom_api] TTS调用异常: {e}", exc_info=True)
            empty_audio = np.zeros(22050, dtype=np.float32)
            return empty_audio, "error.wav"

    async def generate_audio_auto(self, text: str) -> tuple[np.ndarray, str]:
        if hasattr(self, 'config') and not self.config['tts'].get('enable', True):
            logger.info("TTS全局开关关闭，跳过自动TTS接口")
            return None, ""
        """根据配置自动选择TTS接口"""
        if hasattr(self, 'config') and not self.config['tts'].get('enable', True):
            if not getattr(self, '_tts_disable_logged', False):
                logger.info("TTS全局开关关闭，跳过语音合成")
                self._tts_disable_logged = True
            return None, ""
        engine = self.get_engine()
        if engine in ('gpt-sovits', 'sovits'):
            return await self.generate_audio(text)
        elif engine in ('vits', 'vits-simple-api'):
            return await self.vits_simple_api_generate_audio(text)
        elif engine in ('huahua', 'hua-hua-api'):
            return await self.hua_hua_api_generate_audio(text)
        elif engine == 'custom':
            return await self.custom_api_generate_audio(text)
        else:
            logger.warning(f"未知TTS引擎: {engine}，使用默认gpt-sovits")
            return await self.generate_audio(text)

    async def vits_simple_api_generate_audio(self, text: str) -> tuple[np.ndarray, str]:
        """使用vits-simple-api接口生成语音音频"""
        # logger.debug(f"[TTS调用] vits_simple_api_generate_audio 参数 text: {repr(text)}")
        if not text:
            logger.warning("空文本，生成静音音频")
            empty_audio = np.zeros(22050, dtype=np.float32)
            return empty_audio, "empty.wav"

        # 检查缓存
        cache_file = os.path.join(self.cache_dir, f"vits_simple_{hash(text)}.wav")
        if os.path.exists(cache_file):
            logger.info(f"[vits_simple_api] 使用缓存音频: {cache_file}")
            try:
                data, samplerate = sf.read(cache_file)
                return data, cache_file
            except Exception as e:
                logger.error(f"[vits_simple_api] 读取缓存音频失败: {e}")

        # 生成新音频
        try:
            url = self.vits_url if hasattr(self, 'vits_url') else ""
            if not url or not url.startswith('http'):
                logger.error("[vits_simple_api] TTS接口URL未配置或非法，请在配置中填写正确的vits_url！")
                empty_audio = np.zeros(22050, dtype=np.float32)
                return empty_audio, "error.wav"
            params = {
                "text": text,
                "id": 144,
                "format": "wav",
                "lang": "auto",
                "length": 1,
                "noise": 0.33,
                "noisew": 0.4,
                "max": 50
            }
            start_time = time.time()
            response = await asyncio.to_thread(
                requests.get,
                url,
                params=params,
                timeout=30
            )
            if response.status_code == 200:
                with open(cache_file, "wb") as f:
                    f.write(response.content)
                data, samplerate = sf.read(cache_file)
                logger.info(f"[vits_simple_api] 音频生成成功，耗时: {time.time() - start_time:.2f}s")
                return data, cache_file
            else:
                logger.error(f"[vits_simple_api] TTS服务错误: {response.status_code} - {response.text}")
                empty_audio = np.zeros(22050, dtype=np.float32)
                return empty_audio, "error.wav"
        except Exception as e:
            logger.error(f"[vits_simple_api] TTS调用异常: {e}", exc_info=True)
            empty_audio = np.zeros(22050, dtype=np.float32)
            return empty_audio, "error.wav"
    async def generate_audio_and_play(self, text):
        if hasattr(self, 'config') and not self.config['tts'].get('enable', True):
            if not getattr(self, '_tts_disable_logged', False):
                logger.info("TTS全局开关关闭，跳过语音合成")
                self._tts_disable_logged = True
            return None, ""
        import pygame
        audio_data, audio_path = await self.generate_audio_auto(text)
        if audio_data is not None and os.path.exists(audio_path):
            pygame.mixer.Sound(audio_path).play()
        else:
            logger.warning("语音生成失败，无法播放音频")
    """语音合成引擎，将文本转换为语音"""

    def __init__(self, config: Dict):
        self.config = config
        self.endpoint = config["tts"].get("sovits_url", "")
        self.cache_dir = config["tts"].get("audio_cache_dir", "data/cache")
        self.engine = config["tts"].get("engine", "gpt-sovits")
        self.vits_url = config["tts"].get("vits_url", "")
        self.huahua_url = config["tts"].get("huahua_url", "")
        self.custom_url = config["tts"].get("custom_url", "")
        self.audio_cache_max_files = config["tts"].get("audio_cache_max_files", 30)
        os.makedirs(self.cache_dir, exist_ok=True)
        logger.info(f"TTS引擎初始化，缓存目录: {self.cache_dir}，engine: {self.engine}，max_cache: {self.audio_cache_max_files}")

    async def generate_audio(self, text: str) -> tuple[np.ndarray, str]:
        """异步生成语音音频"""
        # logger.debug(f"[TTS调用] generate_audio 参数 text: {repr(text)}")
        if not text:
            logger.warning("空文本，生成静音音频")
            empty_audio = np.zeros(22050, dtype=np.float32)
            return empty_audio, "empty.wav"

        # 检查缓存
        cache_file = os.path.join(self.cache_dir, f"{hash(text)}.wav")
        if os.path.exists(cache_file):
            logger.info(f"使用缓存音频: {cache_file}")
            try:
                data, samplerate = sf.read(cache_file)
                return data, cache_file
            except Exception as e:
                logger.error(f"读取缓存音频失败: {e}")
        # 缓存清理
        self._cleanup_cache()

        # 生成新音频
        try:
            if not self.endpoint or not self.endpoint.startswith('http'):
                logger.error("[gpt-sovits] TTS接口URL未配置或非法，请在配置中填写正确的sovits_url！")
                empty_audio = np.zeros(22050, dtype=np.float32)
                return empty_audio, "error.wav"
            payload = {
                "text": text,
                "text_language": "zh",
                "ref_audio_path": ""  # 可根据需要添加参考音频
            }

            start_time = time.time()
            # 解析 endpoint 和 params
            import urllib.parse
            url, _, query = self.endpoint.partition('?')
            params = dict(urllib.parse.parse_qsl(query))
            # 用 text 替换 {text} 占位符，且只替换一次，防止AI fallback内容覆盖
            replaced = False
            for k, v in params.items():
                if v == "{text}" and not replaced:
                    params[k] = text
                    replaced = True
            response = await asyncio.to_thread(
                requests.get,
                url,
                params=params,
                timeout=30
            )

            if response.status_code == 200:
                # 保存音频到缓存
                with open(cache_file, "wb") as f:
                    f.write(response.content)

                # 读取音频数据
                data, samplerate = sf.read(cache_file)
                logger.info(f"音频生成成功，耗时: {time.time() - start_time:.2f}s")
                return data, cache_file
            else:
                logger.error(f"TTS服务错误: {response.status_code} - {response.text}")
                empty_audio = np.zeros(22050, dtype=np.float32)
                return empty_audio, "error.wav"

        except Exception as e:
            logger.error(f"TTS调用异常: {e}", exc_info=True)
            empty_audio = np.zeros(22050, dtype=np.float32)
            return empty_audio, "error.wav"